package mqtt

import (
	"encoding/json"
	"errors"
	"fmt"
	"fsdvgw/config"
	"fsdvgw/repo"
	"fsdvgw/support_device"
	mqtt "github.com/eclipse/paho.mqtt.golang"
	"log"
	"time"
)

var client mqtt.Client

var messageHandler mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
	receiveEntity := &CommandEntity{}
	err := json.Unmarshal(msg.Payload(), receiveEntity)
	if err != nil {
		//多网关的情况下，非目标网关此处reply会有问题
		reply(receiveEntity.RequestId, err)
		return
	}
	if receiveEntity.RequestId == "" {
		log.Println("消息缺少requestId")
		return
	}
	if receiveEntity.Id == "" || receiveEntity.Command == "" {
		reply(receiveEntity.RequestId, errors.New("消息缺少设备id或命令字符"))
		return
	}

	//这里收到的id是和das保持一致的主键ID
	device, exist, err := repo.GetDeviceByPrimaryId(receiveEntity.Id)
	if err != nil {
		//多网关的情况下，非目标网关此处reply会有问题
		reply(receiveEntity.RequestId, err)
		return
	}
	//如果找不到设备，不做任何响应
	if !exist {
		return
	}

	result, err := support_device.ExecuteCommand(device, receiveEntity.Command, receiveEntity.Body)
	if err != nil {
		reply(receiveEntity.RequestId, err)
		return
	}
	reply(receiveEntity.RequestId, result)
}

var connectHandler mqtt.OnConnectHandler = func(client mqtt.Client) {
	if token := client.Subscribe(config.GetConf().OrgId, 0, messageHandler); token.Wait() && token.Error() != nil {
		panic(token.Error())
	}
	log.Println("Connected Mqtt Server")
}

var connectLostHandler mqtt.ConnectionLostHandler = func(client mqtt.Client, err error) {
	log.Println("Mqtt Server Connection Lost", err)
}

func Start() {
	options := mqtt.NewClientOptions()
	address := fmt.Sprintf("tcp://%s:%d", config.GetConf().MqttAddress, config.GetConf().MqttPort)
	options.AddBroker(address)
	options.SetClientID("device_gateway_" + config.GetConf().OrgId)
	//options.SetUsername(config.GetConf().UserName)
	//options.SetPassword(config.GetConf().Password)
	options.AutoReconnect = true
	options.CleanSession = false
	options.OnConnect = connectHandler
	options.OnConnectionLost = connectLostHandler
	client = mqtt.NewClient(options)
	if token := client.Connect(); token.Wait() && token.Error() != nil {
		panic(token.Error())
	}

}

func Stop() {
	client.Unsubscribe(config.GetConf().OrgId)
	client.Disconnect(2000)
}

func reply(requestId string, result any) {
	replyTopic := config.GetConf().OrgId + "/" + requestId

	var entity *ReplyEntity
	err, ok := result.(error)
	if ok {
		entity = &ReplyEntity{
			RequestId: requestId,
			Success:   false,
			Msg:       err.Error(),
			Data:      nil,
		}
	} else {
		entity = &ReplyEntity{
			RequestId: requestId,
			Success:   true,
			Msg:       "",
			Data:      result,
		}
	}
	bytes, err := json.Marshal(entity)
	if err != nil {
		log.Println(err)
	}
	//发送回复qos设置为0的原因：付过qos设置为2,网关处理消息的时间如果超过服务端等待的超时时间，服务端已不再订阅，而网关会无限reply
	token := client.Publish(replyTopic, 0, true, bytes)

	if !token.WaitTimeout(5 * time.Second) {
		log.Println("reply超时")
	}
	if token.Error() != nil {
		log.Println("reply异常:" + token.Error().Error())
	}
}

type CommandEntity struct {
	Id        string         `json:"id"`
	RequestId string         `json:"requestId"`
	Command   string         `json:"command"`
	Body      map[string]any `json:"body"`
}

type ReplyEntity struct {
	RequestId string `json:"requestId"`
	Success   bool   `json:"success"`
	Msg       string `json:"msg"`
	Data      any    `json:"data"`
}
